/*
* This file is a part of the open-eBackup project.
* This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
* If a copy of the MPL was not distributed with this file, You can obtain one at
* http://mozilla.org/MPL/2.0/.
*
* Copyright (c) [2024] Huawei Technologies Co.,Ltd.
*
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
*/
#ifndef OBJECT_STORAGE_BASE_SERVICE_H
#define OBJECT_STORAGE_BASE_SERVICE_H
#define DEVSTUB

/* System Header files */
#include <atomic>
#include <map>
#include <memory>
#include <arpa/inet.h>
#include <tuple>

/* Module Header files */
#include "define/Types.h"
#include "common/Thread.h"
#include "common/JsonHelper.h"
#include "common/Path.h"
#include "common/CleanMemPwd.h"
#include "common/Snowflake.h"
#include "device_access/DeviceManager.h"
#include "device_access/ControlDevice.h"

/* Plugin Framework Header files */
#include "thrift_interface/ApplicationProtectBaseDataType_types.h"
#include "thrift_interface/ApplicationProtectPlugin_types.h"
#include "thrift_interface/ApplicationProtectFramework_types.h"
#include "job/BasicJob.h"
#include "job/JobCommonInfo.h"
#include "job_mgr/JobMgr.h"
#include "client/ClientInvoke.h"

/* Obs Service Header files */
#include "common/constant/ErrorCode.h"
#include "common/constant/PluginGeneralTypes.h"
#include "common/constant/PluginConstants.h"
#include "utils/PluginUtilities.h"

/* Scanner module */
#include "model/ScanConfig.h"

/* Scanner "ObjectStorageCommonService" */
#include "ScanMgr.h"

/* Backup module */
#include "BackupMgr.h"

/* Common Struct */
#include "CommonStruct.h"
#include "JsonFileTool.h"
#include "ObjectConfigManager.h"

namespace OBSPlugin {

struct TaskInfo {
    int64_t mLastLogReportTime = 0;         /* Last when we reported report to PM */
};

// used to track agent UBC rpc call duration
template<typename T, typename F1, typename F2>
void InterfaceInvokeTimer(std::chrono::duration<T> timeout, F1 procedure, F2 timeoutCallback)
{
    auto begin = std::chrono::steady_clock::now();
    procedure();
    auto end = std::chrono::steady_clock::now();
    if (end - begin >= timeout) {
        timeoutCallback();
    }
}

class NodeLevelTaskInfo {
public:
    void Insert(const std::string mainTaskId)
    {
        HCP_Log(DEBUG, MODULECOM_OBS_COMMON) << "Enter  Insert" << HCPENDLOG;
        struct TaskInfo info{};
        std::lock_guard<std::shared_mutex> lck(m_mtxLock);
        if (CheckKeyExist(mainTaskId)) {
            return;
        }
        m_mainTaskMap.emplace(mainTaskId, info);
        HCP_Log(DEBUG, MODULECOM_OBS_COMMON) << "Exit Insert" << HCPENDLOG;
        return;
    }

    void Erase(std::string mainTaskId)
    {
        HCP_Log(DEBUG, MODULECOM_OBS_COMMON) << "Enter Erase" << HCPENDLOG;
        std::lock_guard<std::shared_mutex> lck(m_mtxLock);
        if (!CheckKeyExist(mainTaskId)) {
            return;
        }
        m_mainTaskMap.erase(mainTaskId);
        HCP_Log(DEBUG, MODULECOM_OBS_COMMON) << "Exit Erase" << HCPENDLOG;
        return;
    }

    void Clear(const std::string mainTaskId)
    {
        DBGLOG("Clear task %s", mainTaskId.c_str());
        std::lock_guard<std::shared_mutex> lck(m_mtxLock);
        m_mainTaskMap.clear();
    }
    bool CheckKeyExist(const std::string &key)
    {
        return m_mainTaskMap.count(key) != 0;
    }
    int GetTaskCount()
    {
        return m_mainTaskMap.size();
    }

    void IncrSubTasksCount()
    {
        HCP_Log(DEBUG, MODULECOM_OBS_COMMON) << "Enter IncrSubTasksCount" << HCPENDLOG;
        m_noOfCurrRunningSubTasks++;
        m_noOfTotalSubTasks++;
    }

    void DecrSubTasksCount()
    {
        HCP_Log(DEBUG, MODULECOM_OBS_COMMON) << "Enter DecrSubTasksCount" << HCPENDLOG;
        if (m_noOfCurrRunningSubTasks != 0) {
            m_noOfCurrRunningSubTasks--;
        }
    }

    uint64_t GetSubTasksCount()
    {
        HCP_Log(DEBUG, MODULECOM_OBS_COMMON) << "Enter GetSubTasksCount" << HCPENDLOG;
        return m_noOfTotalSubTasks;
    }

    uint64_t GetRunningSubTasksCount()
    {
        HCP_Log(DEBUG, MODULECOM_OBS_COMMON) << "Enter GetRunningSubTasksCount" << HCPENDLOG;
        return m_noOfCurrRunningSubTasks;
    }

    bool CanSendLogReportToPM(const std::string mainTaskId)
    {
        HCP_Log(DEBUG, MODULECOM_OBS_COMMON) << "Enter CanSendLogReportToPM" << HCPENDLOG;
        std::lock_guard<std::shared_mutex> lck(m_mtxLock);
        // 120s，单控的任务上报时间间隔
        if ((OBSPlugin::GetCurrentTimeInSeconds() - m_mainTaskMap[mainTaskId].mLastLogReportTime) <
            OBSPlugin::BACKUP_REPORT_CIRCLE_TIME_IN_SEC) {
            HCP_Log(DEBUG, MODULECOM_OBS_COMMON) << "Exit CanSendLogReportToPM ret false" << HCPENDLOG;
            return false;
        }
        HCP_Log(DEBUG, MODULECOM_OBS_COMMON) << "Exit CanSendLogReportToPM ret true" << HCPENDLOG;
        m_mainTaskMap[mainTaskId].mLastLogReportTime = OBSPlugin::GetCurrentTimeInSeconds();
        return true;
    }

    bool UpdateLogReportTimeToPM(const std::string mainTaskId)
    {
        std::lock_guard<std::shared_mutex> lck(m_mtxLock);
        m_mainTaskMap[mainTaskId].mLastLogReportTime = OBSPlugin::GetCurrentTimeInSeconds();
        return true;
    }

    void IncrSubTasksSubmitCount()
    {
        HCP_Log(DEBUG, MODULECOM_OBS_COMMON) << "Enter IncrSubTasksSubmitCount" << HCPENDLOG;
        m_noOfSubTasksGenerated++;
    }

    uint64_t GetSubTasksSubmitCount()
    {
        HCP_Log(DEBUG, MODULECOM_OBS_COMMON) << "Enter GetRunningSubTasksCount" << HCPENDLOG;
        return m_noOfSubTasksGenerated;
    }

private:
    std::shared_mutex m_mtxLock {};                     /* mutex lock */
    std::map<std::string, struct TaskInfo> m_mainTaskMap {};  /* node+main-task-id level info */
    std::atomic<uint64_t> m_noOfCurrRunningSubTasks = 0;    /* node level info - no of sub-tasks currently running */
    std::atomic<uint64_t> m_noOfTotalSubTasks = 0;     /* node level info - total no of sub-tasks (rung + cmpltd) */
    std::atomic<uint64_t> m_noOfSubTasksGenerated = 0;
};

class ObjectStorageCommonService : public BasicJob {
public:
    ObjectStorageCommonService();
    virtual ~ObjectStorageCommonService() {};
    virtual int PrerequisiteJob() = 0;
    virtual int GenerateSubJob() = 0;
    virtual int ExecuteSubJob() = 0;
    virtual int PostJob() = 0;

    bool GetSubJobTypeByFileName(const std::string &fileName, std::string &subTaskName,
                                            uint32_t &subTaskType, uint32_t &subTaskPrio);

    bool InitSubTask(SubJob &subJob, const std::string& jobId,
        const std::string& ctrlFile, uint32_t &subTaskType, const std::string& ext);

    bool IsValidCtrlFile(uint32_t subTaskType, std::string ctrlFileFullPath);

    /**
     * API to create a sub-task
     */
    int CreateSubTask(std::vector<SubJob> &subJobList, std::vector<std::string> &ctrlFileList, bool isCheckPoint);
    int CreateSubTask(const SubJob &job);

    /**
     * API to Report job logs to DME_UBC/PM
     */
    bool ReportJobDetails(const SubJobDetails &jobInfo);

    /**
     * APIs to create/get/update/delete shared resource to store genetal info of a job
     */
    bool LockGeneralResource(std::string mainJobId);
    bool UnlockGeneralResource(std::string mainJobId);
    bool CreateGeneralResource(std::string key, ObjectStorageNativeGeneral &generalInfo);
    bool GetGeneralResource(std::string key, ObjectStorageNativeGeneral &generalInfo);
    bool UpdateGeneralResource(std::string key, ObjectStorageNativeGeneral &generalInfo);
    bool UpdateGeneralResourceStub(std::string key, ObjectStorageNativeGeneral &generalInfo);
    bool DeleteGeneralResource(std::string key);

    /**
     * APIs to create/get/update/delete shared resource to store scanner statistics of a job
     */
    bool LockScanStatsResource(const std::string& mainJobId);
    bool UnlockScanStatsResource(const std::string& mainJobId);
    bool CreateScanStatsResource(std::string key, ObjectStorageNativeScanStatistics &scanStats);
    bool GetScanStatsResource(std::string key, ObjectStorageNativeScanStatistics &scanStats);
    bool UpdateScanStatsResource(std::string key, ObjectStorageNativeScanStatistics &scanStats);
    bool UpdateScanStatsResourceStub(std::string key, ObjectStorageNativeScanStatistics &scanStats);
    bool DeleteScanStatsResource(std::string key);
    bool UpdateScanStatsResourceWithLock(ObjectStorageNativeScanStatistics& scanStatistics,
        bool& bReportToPM);

    /**
     * APIs to create/get/update/delete shared resource to store backup statistics of a job
     */
    bool LockBackupStatsResource(std::string mainJobId);
    bool UnlockBackupStatsResource(std::string mainJobId);
    bool CreateBackupStatsResource(std::string key, ObjectStorageNativeBackupStats &backupStats);
    bool GetBackupStatsResource(std::string key, ObjectStorageNativeBackupStats &backupStats);
    bool UpdateBackupStatsResource(std::string key, ObjectStorageNativeBackupStats &backupStats);
    bool DeleteBackupStatsResource(std::string key);

    /**
     * APIs to get/update/delete shared resource to store backup sub task count of a job
     */
    bool WriteBackupStatsToFile(std::string filePath, ObjectStorageNativeBackupStats &backupStatistics);
    bool ReadBackupStatsFromFile(std::string filePath, ObjectStorageNativeBackupStats &backupStatistics);

    ScanStatistics AddScanStatistics(const ScanStatistics& stats_A, const ScanStatistics& stats_B);

    void SetJobCtrlPhase(std::string jobCtrlPhase);
    uint64_t GetCurrentTimeFromRemoteServer(std::string path);
    void SetMainJobId(const std::string& jobId);
    void SetSubJobId();

    static bool CreateResource(const std::string key, const std::string &val,
        const std::string& mainJobId = "", const std::string& subJobId = "");
    static bool QueryResource(const std::string key, AppProtect::ResourceStatus &ret,
        const std::string& mainJobId = "", const std::string& subJobId = "");
    static bool UpdateResource(const std::string key, Resource &res,
        const std::string& mainJobId = "", const std::string& subJobId = "");
    static bool DeleteResource(const std::string key,
        const std::string& mainJobId = "", const std::string& subJobId = "");
    static bool LockResource(const std::string key, const uint32_t maxRetryCount, const std::string& mainJobId);
    static bool UnlockResource(const std::string key, const uint32_t maxRetryCount, const std::string& mainJobId);

    bool CreateSharedResources(std::string jobId, ObjectStorageNativeGeneral &generalInfo,
        ObjectStorageNativeScanStatistics &scanStats, ObjectStorageNativeBackupStats &backupStats);

    bool GetSharedResources(std::string jobId, ObjectStorageNativeGeneral &generalInfo,
        ObjectStorageNativeScanStatistics &scanStats, ObjectStorageNativeBackupStats &backupStats);

    bool DeleteSharedResources(std::string jobId);

    bool UpdateScannerStatistics(const ObjectStorageNativeScanStatistics& scanStats,
        ObjectStorageNativeScanStatistics& scanStatistics);

    void PrintBackupStatistics(BackupStats &backupStatistics,
        std::string &jobId, BackupPhaseStatus m_backupStatus, std::string &statPrintSubJobId);

    bool UpdateBackupSubTaskStatistics(
        BackupStats &backupStatistics, std::string &metaFsPath, time_t backupStartTime, time_t backupEndTime);

    void PrintSubJobInfo(std::shared_ptr<SubJob> &m_subJobInfo);

    bool IsUpdateBackupStartTimeRequired(ObjectStorageBackupSubJob& backupSubJob,
        ObjectStorageNativeGeneral &generalInfo);

    bool PrintFinalBackupStats(ObjectStorageNativeBackupStats &backupStatistics);

    bool PrintFinalScannerStats(ObjectStorageNativeScanStatistics &scanStats);

    void PrintBackupScanFilter(ScanDirectoryFilter dirFilter, ScanFileFilter fileFilter);

    void FillScanConfigForNative(ScanConfig &scanConfig);

    bool checkFilePathAndGetSrcFileList(std::string srcDir, std::string dstDir,
        std::vector<std::string> &srcFileList);

    static void KeepJobAlive(const std::string& mainJobId, const std::string& subJobId);

    void SleepForCreateSubTaskError();
    bool SendJobReportForAliveness();
    bool CheckAndRetryCreateSubTask(std::vector<SubJob> &subJobList,
        std::vector<std::string> &ctrlFileList, int64_t &createJobErrTime, bool isFinal, bool isCheckPoint);

    void MergeBackupFailureRecords();
    bool UpdateScannerSubTaskStatistics(const PrefixInfo& prefixInfo,
        ObjectStorageNativeScanStatistics& scanStatistics,
        const std::string &cacheFsPath, const std::string& subJobId);
    void RemoveScanStatFile(const PrefixInfo& prefixInfo, const std::string& subJobId,
        const std::string &cacheFsPath);
    bool GetBackupScanStats(const std::string& scanStatPath, ObjectStorageNativeScanStatistics& scanStatistics);
    bool CalcSubScanStats(const std::string& scanStatPath,
        bool isTearDown, ObjectStorageNativeScanStatistics& scanStatistics);
    bool CreateSubTaskWithRetry(uint32_t subTaskType,
        const std::string& jobName, uint32_t jobPriority, bool ignoreFailed = true);
    bool CreateZeroBackupJudgeTask(uint32_t jobPriority);
    int ExecuteZeroBackupJudgeSubJob(const std::string& backupStatsFilePath);
    void KeepSubJobAlive(bool& keepAlive);
    bool InitIdGenerator();
    int CopyUnCompressMeta(const std::string& srcDir, const std::string& targetDir);

    bool CheckEncode(const std::string &endPoint, const std::string &bucketName,
        std::unordered_map<std::string, std::vector<std::string>>& bucketPrefixMap, bool &encodeSwitch);
    void ReadEncodeBucket(std::unordered_map<std::string, std::vector<std::string>>& bucketPrefixMap, bool &encodeSwitch);
    std::string GetConfigStrKey(const std::string& sectionName, const std::string& keyName);
    int GetConfigIntKey(const std::string& sectionName, const std::string& keyName);

    static NodeLevelTaskInfo m_nodeLevelTaskInfo;
protected:
    uint64_t m_curProcess = 0;
    std::string m_subJobId;
    std::atomic<int> m_jobProgress {0};
    std::atomic<int64_t> m_dataSize {0};
    std::atomic<int32_t> m_averageSpeed {0};
    std::string m_jobCtrlPhase;
    ObjectConfigManager m_objectConfigManager;

    /* Scanner instance */
    std::unique_ptr<Scanner> m_scanner { nullptr };
    bool m_isScanParallel = false; // 是否支持扫描并行: false-不支持，true-支持并行
    ScanMgr& m_scanMgr;

    /* Subtask completion status of scan task */
    SCANNER_STATUS m_scanStatus {};
 
    int64_t m_lastJobReportTime {0};
 
    /* Root Path Of Backup Failure Recorder Output */
    std::string m_failureRecordRoot {};
    uint64_t m_maxFailureRecordsNum = 0;

    std::shared_ptr<Module::Snowflake> m_idGenerator { nullptr };
 
    /* access info */
    ObjectStorageAuthExtendInfo m_authInfo;

    std::string m_certFilePath;
 
    void PrintScannerStatistics(ObjectStorageNativeScanStatistics &scanStatistics);
 
    template <typename... Args>
    void
    ReportJobDetailsWithLabelAndExtendInfo(
        const std::tuple<JobLogLevel::type, SubJobStatus::type, const int> &reportInfo,
        std::string logLabel, int64_t errCode,
        const std::string& extendInfo, Args... logArgs)
    {
        SubJobDetails subJobDetails;
        LogDetail logDetail{};
        std::vector<LogDetail> logDetailList;
        ActionResult result;
        int32_t jobSpeed = 0;
        SubJobStatus::type jobStatus = std::get<1>(reportInfo);
        if (IsAbortJob() && jobStatus == SubJobStatus::RUNNING) {
            INFOLOG("Job is aborted, force change jobStatus to aborting");
            jobStatus = SubJobStatus::ABORTING;
            logLabel = "";
        }
        AddLogDetail(logDetail, logLabel, std::get<0>(reportInfo), logArgs...);
        AddErrCode(logDetail, errCode);
        if (m_dataSize >= 0) {
            subJobDetails.__set_dataSize(m_dataSize);
            INFOLOG("Report data size of job %s, dataSize: %llu", m_jobId.c_str(), m_dataSize.load());
        }
        if (m_averageSpeed != 0) {
            subJobDetails.__set_speed(m_averageSpeed);
            INFOLOG("Report average speed %d of job %s", m_averageSpeed.load(), m_jobId.c_str());
        }
        if (!extendInfo.empty()) {
            subJobDetails.__set_extendInfo(extendInfo);
        }
        
        INFOLOG("Enter ReportJobDetails: jobId: %s, subJobId: %s, jobStatus: %d, logLabel: %s, errCode: %d",
                m_jobId.c_str(), m_subJobId.c_str(), static_cast<int>(jobStatus), logLabel.c_str(), errCode);
        constexpr int nuM2 = 2;
        REPORT_LOG2AGENT(subJobDetails, result, logDetailList, logDetail, std::get<nuM2>(reportInfo), jobSpeed,
                         jobStatus);
        if (result.code != Module::SUCCESS) {
            ERRLOG("ReportJobDetails failed, jobId: %s, subJobId: %s, errCode: %d", m_jobId.c_str(), m_subJobId.c_str(),
                   result.code);
        }
 
        INFOLOG("Exit ReportJobDetails: jobId: %s, subJobId: %s, jobStatus: %d, logLabel: %s, errCode: %d",
                m_jobId.c_str(), m_subJobId.c_str(), static_cast<int>(jobStatus), logLabel.c_str(), errCode);
        return;
    }

    template <typename... Args>
    void
    ReportJobDetailsWithLabelAndErrcode(const std::tuple<JobLogLevel::type, SubJobStatus::type, const int> &reportInfo,
                                        std::string logLabel, int64_t errCode, Args... logArgs)
    {
        ReportJobDetailsWithLabelAndExtendInfo(reportInfo, logLabel, errCode, "", logArgs...);
    }
 
    template <typename... Args>
    void ReportJobDetails(const JobLogLevel::type &logLevel, SubJobStatus::type jobStatus, int jobProgress,
                          Args... logArgs)
    {
        ReportJobDetailsWithLabelAndErrcode(std::make_tuple(logLevel, jobStatus, jobProgress), "", 0, logArgs...);
    }

    std::string GetScanGenFileRelativeDir(const PrefixInfo& prefixInfo);
    std::string GetScanPrefixHash(const PrefixInfo& prefixInfo);

    template <typename... Args>
    void ReportJobDetailsWithLabel(SubJobStatus::type jobStatus, int32_t jobProgress,
        std::string logLabel, const JobLogLevel::type &logLevel, Args... logArgs)
    {
        ReportJobDetailsWithLabelAndErrcode(std::make_tuple(logLevel, jobStatus, jobProgress), logLabel, 0, logArgs...);
    }
private:
    // utility API to create unique File Name
    std::string CreateUniqueTempFileName(std::string path);
};
}
#endif  // #ifndef OBJECT_STORAGE_BASE_SERVICE_H